package com.hucais.app.ods;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.hucais.bean.staff.ChannelQueryDetail;
import com.hucais.bean.staff.SelectStaffBaseDetail;
import com.hucais.bean.staff.SelectStaffExtDetail;
import com.hucais.common.AppConstant;
import com.hucais.utils.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 选品参谋
 */
public class SelectStaffApp {
    private static final Logger LOG = LoggerFactory.getLogger(SelectStaffApp.class);

    private static final String SOURCE_TOPIC = DefaultPropertiesUtil.get("kafka.topic.staff");
    private static final String GROUP_ID = "select_staff_app_group";
    private static final String CHECK_POINT_PATH_SUF = DefaultPropertiesUtil.get("checkpoint.path.suf.staff");
    private static final OutputTag<SelectStaffExtDetail> extTag = new OutputTag<SelectStaffExtDetail>("ext") {
    };

    public static void main(String[] args) {
        //获取执行环境
        StreamExecutionEnvironment env = AppCommonUtil.getDefaultStreamEnv(false, CHECK_POINT_PATH_SUF, 2);

        try {
            DataStreamSource<String> kafkaStream = env.addSource(KafkaUtil.getDefaultFlinkKafkaConsumer(SOURCE_TOPIC, GROUP_ID));

            // 将每行数据转换为JSON对象,过滤空值
            SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStream.map(JSON::parseObject).filter(jsonObj -> jsonObj.size() > 0);
            //筛选出内容-数据流
            SingleOutputStreamOperator<JSONObject> contentStream = jsonObjStream
                    .filter(jsonObj -> StringUtils.isNotBlank(jsonObj.getString("task_code")) &&
                            StringUtils.isNotBlank(jsonObj.getString("isbn")) &&
                            StringUtils.isNotBlank(jsonObj.getString("acq_date")) &&
                            StringUtils.isNotBlank(jsonObj.getString("acq_time")))
                    .assignTimestampsAndWatermarks(
                            WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                                    .withTimestampAssigner((SerializableTimestampAssigner<JSONObject>) (jsonObj, recordTimestamp) -> DateUtil.parse(jsonObj.getString("acq_time")).getTime())
                    ).name("watermark");
            // 根据task_code + isbn分组
            KeyedStream<JSONObject, String> keyedStream = contentStream
                    .keyBy((KeySelector<JSONObject, String>) jsonObj -> String.format("%s_%s", jsonObj.getString("task_code"), jsonObj.getString("isbn")));
            SingleOutputStreamOperator<SelectStaffBaseDetail> baseResStream = keyedStream.process(new SplitStreamFunction());
            DataStream<SelectStaffExtDetail> extResStream = baseResStream.getSideOutput(extTag);

            // 数据输出
            sinkTable(baseResStream, extResStream);
            // 执行
            env.execute("SelectStaffApp");
        } catch (Exception e) {
            LOG.error("选品参谋-出现异常", e);
        }
    }

    private static void sinkTable(SingleOutputStreamOperator<SelectStaffBaseDetail> baseResStream, DataStream<SelectStaffExtDetail> extResStream) {
        // 基础信息
        String baseSqlTemplate = "INSERT INTO %s ( " +
                "  task_code, isbn, acq_date, acq_time, channel ,book_name, author, publishing_house, publishing_time, brand, store_pricing,  " +
                "  score, grader_cnt, reading_cnt, have_read_cnt, will_read_cnt,short_comments_cnt, comments_cnt, acq_url " +
                ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE  " +
                "acq_time = VALUES(acq_time),channel = VALUES(channel),book_name = VALUES(book_name),author = VALUES(author),publishing_house = VALUES(publishing_house), publishing_time = VALUES(publishing_time), " +
                "brand = VALUES(brand),store_pricing = VALUES(store_pricing),score = VALUES(score), grader_cnt = VALUES(grader_cnt),reading_cnt = VALUES(reading_cnt), " +
                "have_read_cnt = VALUES(have_read_cnt),will_read_cnt = VALUES(will_read_cnt),short_comments_cnt = VALUES(short_comments_cnt),comments_cnt = VALUES(comments_cnt),acq_url = VALUES(acq_url)";
        String baseSql = String.format(baseSqlTemplate, DefaultPropertiesUtil.get("mysql.table.staff.base"));
        // 扩展信息
        String extSqlTemplate = "INSERT INTO %s (task_code,isbn,acq_date,channel,acq_time,min_selling_price,top3_deatail,top_deatail,detail) " +
                "VALUES (?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE " +
                "acq_time = VALUES(acq_time), min_selling_price= VALUES(min_selling_price),top3_deatail= VALUES(top3_deatail),top_deatail= VALUES(top_deatail),detail= VALUES(detail)";
        String extSql = String.format(extSqlTemplate, DefaultPropertiesUtil.get("mysql.table.staff.ext"));

        baseResStream.addSink(JdbcUtil.getMysqlSinkFunc(baseSql)).name("sinkToBase").setParallelism(1);
        extResStream.addSink(JdbcUtil.getMysqlSinkFunc(extSql)).name("sinkToExt").setParallelism(1);
    }

    /**
     * 拆分多支流处理
     */
    public static class SplitStreamFunction extends ProcessFunction<JSONObject, SelectStaffBaseDetail> {
        @Override
        public void processElement(JSONObject jsonObj, Context ctx, Collector<SelectStaffBaseDetail> out) {
            String taskCode = jsonObj.getString("task_code");
            String isbn = jsonObj.getString("isbn");
            String acqDate = jsonObj.getString("acq_date");
            String acqTime = jsonObj.getString("acq_time");


            if (jsonObj.containsKey(AppConstant.CHANNEL_DOUBAN)) {
                SelectStaffBaseDetail base = new SelectStaffBaseDetail(taskCode, isbn, acqDate, acqTime, AppConstant.CHANNEL_DOUBAN);
                //组装内容-属性
                JSONObject douban = jsonObj.getJSONObject("douban");
                base.setBook_name(douban.getString("book_name"));
                base.setAuthor(CommonUtil.handleBlankStr(douban.getString("author")));
                base.setPublishing_house(CommonUtil.handleBlankStr(douban.getString("publishing_house")));
                base.setPublishing_time(CommonUtil.handleBlankStr(douban.getString("publishing_time")));
                base.setBrand(CommonUtil.handleBlankStr(douban.getString("brand")));
                base.setStore_pricing(CommonUtil.handleBlankDouble(douban.getString("store_pricing")));
                base.setScore(CommonUtil.handleBlankDouble(douban.getString("score")));
                base.setGrader_cnt(CommonUtil.handleBlankNumber(douban.getString("grader_cnt")));
                base.setReading_cnt(CommonUtil.handleBlankNumber(douban.getString("reading_cnt")));
                base.setHave_read_cnt(CommonUtil.handleBlankNumber(douban.getString("have_read_cnt")));
                base.setWill_read_cnt(CommonUtil.handleBlankNumber(douban.getString("will_read_cnt")));
                base.setShort_comments_cnt(CommonUtil.handleBlankNumber(douban.getString("short_comments_cnt")));
                base.setComments_cnt(CommonUtil.handleBlankNumber(douban.getString("comments_cnt")));
                base.setAcq_url(CommonUtil.handleBlankStr(douban.getString("acq_url")));
                out.collect(base);
            } else if (jsonObj.containsKey(AppConstant.CHANNEL_JD)) {
                JSONArray channelDetail = jsonObj.getJSONArray(AppConstant.CHANNEL_JD);
                JSONObject topDetail = jsonObj.getJSONObject(AppConstant.JSON_KEY_TOP);
                SelectStaffExtDetail ext = new SelectStaffExtDetail(taskCode, isbn, acqDate, AppConstant.CHANNEL_JD, acqTime);
                processData(ext, AppConstant.CHANNEL_JD, topDetail, channelDetail);
                ctx.output(extTag, ext);
            } else if (jsonObj.containsKey(AppConstant.CHANNEL_TB)) {
                JSONArray channelDetail = jsonObj.getJSONArray(AppConstant.CHANNEL_TB);
                JSONObject topDetail = jsonObj.getJSONObject(AppConstant.JSON_KEY_TOP);
                SelectStaffExtDetail ext = new SelectStaffExtDetail(taskCode, isbn, acqDate, AppConstant.CHANNEL_TB, acqTime);
                processData(ext, AppConstant.CHANNEL_TB, topDetail, channelDetail);
                ctx.output(extTag, ext);
            } else if (jsonObj.containsKey(AppConstant.CHANNEL_DD)) {
                JSONArray channelDetail = jsonObj.getJSONArray(AppConstant.CHANNEL_DD);
                JSONObject topDetail = jsonObj.getJSONObject(AppConstant.JSON_KEY_TOP);
                SelectStaffExtDetail ext = new SelectStaffExtDetail(taskCode, isbn, acqDate, AppConstant.CHANNEL_DD, acqTime);
                processData(ext, AppConstant.CHANNEL_DD, topDetail, channelDetail);
                ctx.output(extTag, ext);
            }

        }

        private void processData(SelectStaffExtDetail detail, String channel, JSONObject topDetail, JSONArray channelDetail) {
            // 筛选出销量Top3并相关处理
            List<ChannelQueryDetail> top3List = getTop3(channelDetail);
            String top3JsonArr = null;
            if (null != top3List) {
                JSONArray top3Arr = new JSONArray();
                for (ChannelQueryDetail queryDetail : top3List) {
                    JSONObject json = new JSONObject();
                    json.put("selling_price", queryDetail.getSellingPrice());
                    json.put("comments_cnt", queryDetail.getCommentsCnt());
                    json.put("acq_url", queryDetail.getAcqUrl());
                    json.put("store_name", queryDetail.getStoreName());
                    top3Arr.add(json);
                }

                top3JsonArr = top3Arr.toJSONString();
            }

            // 获取最低售价
            Double minSellingPrice = getMinSellingPrice(top3List, topDetail);

            // 组装对象
            detail.setChannel(channel);
            detail.setMin_selling_price(minSellingPrice);
            detail.setTop3_deatail(top3JsonArr);
            detail.setTop_deatail(JSON.toJSONString(topDetail));
            detail.setDetail(JSON.toJSONString(channelDetail));
        }

        /**
         * 获取页面综合排序-前3
         *
         * @param channelDetail 页面综合排序详情
         * @return 页面综合排序-销量Top3的详情
         */
        private List<ChannelQueryDetail> getTop3(JSONArray channelDetail) {
            if (CollectionUtil.isNotEmpty(channelDetail)) {
                return channelDetail.stream().map(json -> {
                    JSONObject jsonObject = (JSONObject) json;
                    ChannelQueryDetail c = new ChannelQueryDetail();
                    c.setSellingPrice(CommonUtil.handleBlankDouble(jsonObject.getString("selling_price")));
                    c.setCommentsCnt(CommonUtil.handleBlankNumber(jsonObject.getString("comments_cnt")));
                    c.setAcqUrl(CommonUtil.handleBlankStr(jsonObject.getString("acq_url")));
                    c.setStoreName(CommonUtil.handleBlankStr(jsonObject.getString("store_name")));
                    return c;
                }).sorted(Comparator.comparing(ChannelQueryDetail::getCommentsCnt).reversed()).limit(3).collect(Collectors.toList());
            }
            return null;
        }

        /**
         * 获取最低售价
         *
         * @param top3List  页面综合排序-销量Top3的详情
         * @param topDetail 页面销量排序-最高的详情
         * @return 最低售价
         */
        private Double getMinSellingPrice(List<ChannelQueryDetail> top3List, JSONObject topDetail) {
            // 提取销量最高的售价
            Double topSellingPrice = 0.00d;
            if (!topDetail.isEmpty()) {
                topSellingPrice = CommonUtil.handleBlankDouble(topDetail.getString("top_selling_price"));
            }

            // 从Top3和销量最高计算最低售价
            List<Double> sellingPriceList;
            if (CollectionUtil.isNotEmpty(top3List)) {
                sellingPriceList = top3List.stream().map(ChannelQueryDetail::getSellingPrice).collect(Collectors.toList());
                sellingPriceList.add(topSellingPrice);
                return CollectionUtil.min(sellingPriceList);
            } else {
                return topSellingPrice;
            }
        }

    }

}
